package com.gugee.ins.data.collector.media.listener;

import com.gugee.ins.data.collector.media.config.InsMediaConfig;
import com.gugee.ins.data.collector.media.service.InsMediaService;
import com.gugee.ins.data.collector.media.service.InsProductService;
import com.gugee.ins.data.common.consumer.AbstractGugeeConsumerHandler;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 帖子收入监听器
 * @author Huangxin
 */
@Component
@Slf4j
public class InsMediaCollectListener {
    /**
     * 配置类
     */
    @Autowired
    InsMediaConfig insMediaConfig;

    /**
     * InsMedia服务
     */
    @Autowired
    InsMediaService insMediaService;

    /**
     * InsMedia服务
     */
    @Autowired
    InsProductService insProductService;


    private InsMediaCollectHandler collectHandler = new InsMediaCollectHandler();

    private InsProductMediaCollectHandler appClHandler = new InsProductMediaCollectHandler();

    /**
     * 监听任务
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMedia.id}",
            topics = {"${kafka.topics.insMedia.online}","${kafka.topics.insMedia.priority}"},
            containerFactory = "factoryInsMediaAck")
    public void mediaCollect(List<InsBloggerMessage> messages, Acknowledgment ack) {
        collectHandler.handle("InsMediaCollect",messages, insMediaConfig.getAcceptInsMediaMsgTimeMinRange(),ack);
    }

    /**
     * 监听任务
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insAppProductMedia.id}",
            topics = {"${kafka.topics.insAppProductMedia.online}"},
            containerFactory = "factoryInsAppProductMediaAck")
    public void productMediaCollect(List<InsBloggerMessage> messages, Acknowledgment ack) {
        appClHandler.handle("InsAppProductMediaCollect",messages, insMediaConfig.getAcceptInsAppProductMediaMsgTimeMinRange(),ack);
    }

    /**
     * 中播主帖子采集，监听任务
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insMiddleBloggerMedia.id}",
            topics = {"${kafka.topics.insMiddleBloggerMedia.online}"},
            containerFactory = "factoryInsMiddleBloggerMediaAck")
    public void middleBloggerMediaCollect(List<InsBloggerMessage> messages, Acknowledgment ack) {
        collectHandler.handle("InsMiddleBloggerMediaCollect",messages, insMediaConfig.getAcceptInsMiddleBloggerMediaMsgTimeMinRange(),ack);
    }

    class InsMediaCollectHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage> {
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            insMediaService.handleInsMediaCollect(messages);
        }
    }

    class InsProductMediaCollectHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage> {
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            insProductService.handleInsAppProductMediaCl(messages);
        }
    }
}
